package cn._51doit.flink.day10;


import org.apache.avro.generic.GenericRecord;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.avro.ParquetAvroWriters;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.util.concurrent.TimeUnit;

/**
 * 将数据使用Flink的FileSink写入到HDFS中（以parquet格式）
 */
public class FileSinkDemo2 {

    public static void main(String[] args) throws Exception {

        System.setProperty("HADOOP_USER_NAME", "root");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(30000);
        //tom,1
        //jerry,2
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<UserBean> userStream = lines.map(new MapFunction<String, UserBean>() {
            @Override
            public UserBean map(String line) throws Exception {
                String[] fields = line.split(",");
                return new UserBean(fields[0], Integer.parseInt(fields[1]));
            }
        });

        String outputPath = "hdfs://node-1.51doit.cn:9000/out29";

        final FileSink<UserBean> sink = FileSink
                .forBulkFormat(new Path(outputPath), ParquetAvroWriters.forReflectRecord(UserBean.class))
                .build();

        userStream.sinkTo(sink);

        env.execute();

    }

    //封装数据的Bean，这个Bean中就带有了Schema
    public static class UserBean {

        private String name;

        private Integer age;

        public UserBean(){}

        public UserBean(String name, Integer age) {
            this.name = name;
            this.age = age;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public Integer getAge() {
            return age;
        }

        public void setAge(Integer age) {
            this.age = age;
        }
    }
}
